import * as cluster from 'cluster';
import { EventEmitter } from 'events';
import { debuglog } from 'util';

interface Task {
  /** 任务标识，需要保证唯一性 */
  id?: string;
  /** 任务传参 */
  args: any[];
  /** 任务文件的绝对路径 */
  executableFilePath: string;
};

interface Message {
  type: string;
}

/** 子进程日志 */
interface LogMessage extends Message {
  /** 日志信息 */
  args: any[];
}

/** 子进程已成功获取可执行路径，准备好运行 */
interface ReadyMessage extends Message { }

/** 子进程完成分配的任务 */
interface FinishMessage extends Message {
  /** 任务完成耗时 */
  takeUpTime: number;
  /** 任务详情 */
  task: Task;
}

/** 子进程出现错误 */
interface ErrorMessage extends Message {
  /** 任务详情 */
  task: Task;
}
/** 子进程汇报当前状态 */
interface StatusMessage extends Message {
  /** 状态 */
  status: string;
}

interface MessageEvent {
  on(event: 'log', listener: (message: LogMessage, worker: cluster.Worker) => void): void;
  on(event: 'finish', listener: (message: FinishMessage, worker: cluster.Worker) => void): void;
  on(event: 'error', listener: (message: ErrorMessage, worker: cluster.Worker) => void): void;
  on(event: 'ready', listener: (message: ReadyMessage, worker: cluster.Worker) => void): void;
  on(event: 'status', listener: (message: StatusMessage, worker: cluster.Worker) => void): void;
  on(event: string, listener: (...args: any[]) => void): void;
  emit(event: string | symbol, ...args: any[]): boolean;
}

const log = debuglog('master');
const DEFAULT = Symbol('default');

/** 进程池 */
export class ProcessPool {
  /** 工作进程最大数量 */
  public maxProcessCount = 0;

  /** 空闲进程池 */
  public idlePool = new Set<cluster.Worker>();
  
  /** 主进程接受工作进程的message事件 */
  public messageEvent: MessageEvent = new EventEmitter();
  
  /** 任务重试次数 */
  public retryCount = 3;
  
  /** 失败的任务 */
  public errorTaskList: Task[] = [];
  
  /** 控制集群状态，设置为false将会优雅退出 */
  private run = true;

  /** 空闲进程存活时间(单位: 毫秒)，如果设置为0则表示不进行回收 */
  private keepAliveTime: number;

  /** 进程的工种退化期(单位: 毫秒)，如果设置为0则表示不会退化 */
  private deteriorateTime: number;

  /** 标记任务失败次数 */
  private errorTaskMap: { [id: string]: number } = {};

  /** 任务对列 */
  private undealTaskList: Task[] = [];

  /** 空闲进程销毁的定时器 */
  private destroyTimer = new Map<cluster.Worker, NodeJS.Timeout>();

  /**进程退化的定时器 */
  private deteriorateTimer = new Map<cluster.Worker, NodeJS.Timeout>();

  /** 按部门类型给进程分配 */
  private department = new Map<string | symbol,  Set<cluster.Worker>>().set(DEFAULT, new Set());

  /** 每个进程启动时注册自己的工种 */
  private register = new Map<cluster.Worker, string | symbol>();

  /** agent进程 */
  private agentMap = new Map<string, cluster.Worker>();

  /** worker进程 */
  private workerPool = new Set<cluster.Worker>();

  /** 工作进程数 */
  private workCount = 0;

  /**
   * @param maxProcess 工作进程最大数量
   * @param deteriorateTime 工作进程退化周期，(单位: 毫秒)，如果设置为0则表示不退化
   * @param keepAliveTime 空闲进程存活时间(单位: 毫秒)，如果设置为0则表示不进行回收
   */
  constructor(maxProcess: number, deteriorateTime = 3000, keepAliveTime = 10000) {
    if (cluster.isWorker) {
      throw new Error('ClusterWorker must run in master!');
    }

    this.maxProcessCount = maxProcess;
    this.keepAliveTime = keepAliveTime;
    this.deteriorateTime = deteriorateTime;

    cluster.setupMaster({ exec: __filename });

    // 定期检查各进程状态，查出假死原因
    // setInterval(() => {
    //   for (const id in cluster.workers) {
    //     cluster.workers[id].send({ type: 'check' });
    //   }
    // }, 10000);

    cluster.on('online', (worker) => {
      if (worker.agent) {
        worker.send({ type: 'runAgent', executableFilePath: worker.executableFilePath });
      } else {
        this.workerPool.add(worker);
        log(`子进程${ worker.process.pid }启动成功！`);
        // 子进程建立成功后分配任务
        this.assignTask(worker);
      }
    });

    cluster.on('exit', (worker) => {
      if (!worker.agent) {
        this.workerPool.delete(worker);
        this.idlePool.delete(worker);
        this.workCount --;
        if (worker.activeDestroy) {
          log(`工作进程 ${ worker.process.pid } 已退出，当前剩余进程数 ${ this.workerPool.size }`);
          return;
        }
        if (this.workCount < this.maxProcessCount) {
          // 未知异常导致的进程退出，需要重新启动
          this.workCount ++;
          cluster.fork();
        }
      } else {
        // 重启Agent进程
        log(`Agent ${ worker.name } 发生错误，进程重启！`);
        const agentWoker = cluster.fork();
        agentWoker.name = worker.name;
        agentWoker.executableFilePath = worker.executableFilePath;
        agentWoker.agent = true;
        this.agentMap.set(worker.name, agentWoker);
      }
    });

    this.messageEvent.on('ready', (_message, worker) => {
      // 将注册好的进程加入对应的部门
      const type = this.register.get(worker);
      if (!this.department.get(type)) {
        this.department.set(type, new Set())
      }
      this.department.get(type).add(worker);
      // 通知工作进程开始正式工作
      worker.send({ type: 'run' });
    });

    this.messageEvent.on('finish', (message, worker) => {
      if (message.type === 'finish' && message.takeUpTime) {
        log(`任务 ${ message.task.id }完成耗时(${ message.takeUpTime })ms`);
      }
      if (!this.run) {
        log(`${ worker.process.pid }手动安全退出`);
        worker.send({ type: 'end' });
        return;
      }
      const type = this.register.get(worker);
      const task = this.getTask(type);
      if (task) {
        // 找到同工种任务，则直接运行
        worker.send({ type: 'run', task });
        return;
      }
      // 没有同工种任务则进入退化期
      this.deteriorateStage(worker);
    });

    this.messageEvent.on('error', (message, worker) => {
      const task = message.task;
      if (!this.errorTaskMap[task.id]) {
        this.errorTaskMap[task.id] = 0;
      }
      if (this.errorTaskMap[task.id] >= this.retryCount) {
        delete this.errorTaskMap[task.id];
        this.errorTaskList.push(task);
        log(`任务[${ task.id }]重试次数达到上限！`);
      } else {
        this.errorTaskMap[task.id] ++;
        log(`任务[${ task.id }]重回对列，稍后进行第${ this.errorTaskMap[task.id] }次重试`);
        this.addTask(task);
      }
      this.messageEvent.emit('finish', message, worker);
    });

    this.messageEvent.on('status', (message, worker) => {
      log(`${ worker.process.pid }当前状态: ${ message.status }`);
    });

    this.messageEvent.on('agent', (message) =>{
      const worker = this.agentMap.get(message.name);
      if (worker) {
        worker.send({ type: 'receive', data: message.data });
      }
    });

    this.messageEvent.on('worker', (message) =>{
      const data = { type: 'receive', data: message.data };
      for (const worker of this.workerPool) {
        worker.send(data);
      }
    });

    cluster.on('message', (worker, message) => {
      this.messageEvent.emit(message.type, message, worker);
    });
  }

  public addTask(task: Task) {
    if (this.idlePool.size > 0) {
      // 默认工种进程
      let defaultWorker: cluster.Worker;
      for (const worker of this.idlePool) {
        const executableFilePath = this.register.get(worker);
        if (executableFilePath === task.executableFilePath) {
          // 取消进程空闲标记
          this.idlePool.delete(worker);
          // 清除退化定时器
          clearTimeout(this.deteriorateTimer.get(worker));
          this.deteriorateTimer.delete(worker);
          // 清除销毁定时器
          clearTimeout(this.destroyTimer.get(worker));
          this.destroyTimer.delete(worker);
          // 运行任务
          worker.send({ type: 'run', task });
          return;
        }
        if (executableFilePath === DEFAULT) {
          defaultWorker = worker;
        }
      }
      if (defaultWorker) {
        // 清除自动销毁的定时器
        clearTimeout(this.destroyTimer.get(defaultWorker));
        this.destroyTimer.delete(defaultWorker);
        // 取消进程空闲标记
        this.idlePool.delete(defaultWorker);
        // 标记进程工种
        this.register.set(defaultWorker, task.executableFilePath);
        // 分配任务
        defaultWorker.send({ type: 'before', task });
        return;
      }
    }
    this.undealTaskList.push(task);
    if (this.workCount < this.maxProcessCount) {
      this.workCount ++;
      // 如果进程没有超过最大限制数，则创建新的工作进程来处理
      cluster.fork();
    }
  }

  /**
   * 优雅的停止
   * 工作进程会在当前任务完成后进入空闲状态，不再接收新的任务
   */
  public stop() {
    this.run = false;
    for (const worker of this.idlePool) {
      worker.send({ type: 'end' });
    }
    this.idlePool.clear();
  }

  /**
   * 重新分配任务
   */
  public start() {
    this.run = true;
    for (let i = 0; i < this.undealTaskList.length; i ++) {
      if (this.workCount === this.maxProcessCount) {
        break;
      }
      // 如果进程没有超过最大限制数，则创建新的工作进程来处理
      this.workCount ++;
      cluster.fork();
    }
  }

  /**
   * 创建agent进程
   * @param name agent名称
   * @param executableFilePath 任务文件的绝对路径
   */
  public createAgent(name: string, executableFilePath: string) {
    if (this.agentMap.get(name)) {
      throw new Error(`${ name } exists！`);
    }
    const worker =  cluster.fork();
    worker.name = name;
    worker.agent = true;
    worker.executableFilePath = executableFilePath;
    this.agentMap.set(name, worker);
  }

  /**
   * 获取待处理任务数量
   */
  public getUndealTaskCount() {
    return this.undealTaskList.length;
  }

  private assignTask(worker: cluster.Worker) {
    if (this.undealTaskList.length > 0) {
      const type = this.register.get(worker);
      let task: Task;
      if (!type || type === DEFAULT) {
        task = this.undealTaskList.shift();
      } else {
        task = this.getTask(type) || this.undealTaskList.shift();
      }
      // 注册工种
      this.register.set(worker, task.executableFilePath);
      worker.send({ type: 'before', task });
    } else { 
      // 标记进程空闲
      this.idlePool.add(worker);
      // 进入销毁期
      this.destoryStage(worker);
    }
  }

  /**
   * 根据工种查询对应的待处理任务，如果不存在则返回最近的任务
   * @param type 工种
   */
  private getTask(executableFilePath: string | symbol): Task | undefined {
    for (let i = 0; i < this.undealTaskList.length; i ++) {
      const task = this.undealTaskList[i];
      if (task.executableFilePath === executableFilePath) {
        this.undealTaskList.splice(i, 1);
        log(`待分配任务: ${ this.undealTaskList.length }`);
        return task;
      }
    }
  }

  /**
   * 工作进程进入销毁期
   * @param worker 工作进程
   */
  private destoryStage(worker: cluster.Worker) {
    if (this.keepAliveTime) {
      // 创建销毁计时器，空闲进程
      this.destroyTimer.set(worker, setTimeout(() => {
        this.idlePool.delete(worker);
        this.destroyTimer.delete(worker);
        worker.activeDestroy = true;
        worker.send({ type: 'end' });
      }, this.keepAliveTime));
    }
  }

  /**
   * 工作进程进入工种退化期
   * @param worker 工作进程
   */
  private deteriorateStage(worker: cluster.Worker) {
    // 标记进程空闲
    this.idlePool.add(worker);

    if (this.deteriorateTime) {
      // 创建退化计时器
      this.deteriorateTimer.set(worker, setTimeout(() => {
        this.deteriorateTimer.delete(worker);
        log(`${ worker.process.pid }工种退化为default`);
        this.register.set(worker, DEFAULT);
        worker.send({ type: 'clean' });
        if (this.undealTaskList.length > 0) {
          this.idlePool.delete(worker);
          // 分配任务
          this.assignTask(worker);
        } else {
          this.destoryStage(worker);
        }
      }, this.deteriorateTime));
    } else {
      this.destoryStage(worker);
    }
  }
}

export class Worker {
  public status = 'begin';
  constructor(public type: string) {}

  /** 运行前的准备工作 */
  public before() { }

  /** 主启动方法 */
  public run(...args) {
    throw new Error('run: interface not implemented');
  }

  /** 进程工种退化时执行 */
  public after() {

  }

  /** 发送消息给指定Agent进程 */
  public sendToAgent(name: string, data: Object) {
    process.send({ type: 'agent', name, data });
  }

  /** 广播消息给其它工作进程 */
  public sendToWorker(data: Object) {
    process.send({ type: 'worker', data });
  }

  public receiveMessage(data: Object) {}
}

export class Agent {
  public status = 'begin';
  constructor(public name: string) {}

  public run() {
    throw new Error('run: interface not implemented');
  }

  public sendToAgent(name: string, data: Object) {
    process.send({ type: 'agent', name, data });
  }

  public sendToWorker(data: Object) {
    process.send({ type: 'worker', data });
  }

  public receiveMessage(data: Object) {}
}

if (cluster.isWorker) {
  let app: Worker | Agent;
  let task: Task;
  const log = debuglog('worker');
  process.on('message', async (message) => {
    switch (message.type) {
      case 'before':
        task = message.task;
        log(`得到任务: ${ task.id }`);
        app = new (require(task.executableFilePath).default)();
        await (app as Worker).before();
        process.send({ type: 'ready' });
        break;
      case 'run':
        task = message.task || task;
        const beginStamp = Date.now();
        try {
          const args = message.args || task.args;
          await app.run(...args);
          process.send({ type: 'finish', task, takeUpTime: Date.now() - beginStamp });
        } catch (error) {
          process.send({ type: 'error', task });
          log('unkown error', error);
        }
        break;
      case 'runAgent':
        app = new (require(message.executableFilePath).default)();
        await app.run();
        break;
      case 'receive':
        app.receiveMessage(message.data);
        break;
      case 'check':
        process.send({ type: 'status', status: app.status });
        break;
      case 'clean':
        await (app as Worker).after();
        break;
      case 'end':
        process.exit();
      default:
        log('unkonw type', message.type);
        break;
    }
  });
}
